#include "boss.h"
using json = nlohmann::json;

void boss::boss_cycle()
{
    int size = 0;
    Socket conn_sock;
    int accept_sock = -1;
    epoll_event temp;
    connection_t* acceptConn;
    bool is_server_socket_received = false;
    auto logger = core_data.logger.setBasicLogger("boss");
    if(core_data.serverSocket.getSocketfd() < 0){
        //log: invalid serverSocket
        logger->critical("初始化失败：无效的serverSocket");
        std::runtime_error("invalid serverSocket");
        return;
    }
    connection_t* serverConn = core_data.connections.getFreeConnection();
    serverConn->setConnnection(core_data.serverSocket.getSocketfd(), READ_TYPE, nullptr,nullptr, false);
    temp = core_data.epoll.setEpollEvent(core_data.serverSocket.getSocketfd(), EPOLLIN | EPOLLET, (void*)serverConn);
    if(!core_data.epoll.epollAddEvent(temp, core_data.serverSocket.getSocketfd())) //将监听socket添加入epoll中，监听socket设为ET
    {
        logger->critical("添加监听socket失败，原因是：{}", MyLogger::getSysError().c_str());   
        std::runtime_error("添加监听socket失败");
        status = -1;
        return;
    }
    while(status >= 0)
    {
        is_server_socket_received = false;
    	size = core_data.epoll.epollWait();
        if(size == 0){
	    //cout << "empty set" << endl;
	    }
        else if(size < 0){
            this-> status = -1;
            logger->critical("epoll_wait()失败，原因是：{}", MyLogger::getSysError().c_str());
            break;
        }
        else{
            //cout << "from acceptor: 接收到"<< size <<"个新连接" << endl;
            vector<epoll_event>& events = core_data.epoll.getEventList(); 
            {
               // std::lock_guard<std::mutex> lock(core_data.queue_mutex);
                for(int i = 0; i < size; ++i){
                    if(events[i].events & EPOLLIN)
                    {
                        acceptConn = (connection_t*) events[i].data.ptr;
                        if(acceptConn->fd == core_data.serverSocket.getSocketfd()){
                            is_server_socket_received = true;
                        }else{
                            if(acceptConn == nullptr)
                            {
                                logger->warn("接收到数据域为空指针的非侦听socket");
                            }
                            else
                            {
                                //core_data.request_queue.push(events[i]);
                                if(acceptConn->reader.get() != nullptr){
                                    acceptConn->reader.reset();
                                }
                                acceptConn->reader = std::make_shared<X7>(); //如果是原有连接发送的新数据，清空原有的协议状态
                                readEventHandler(acceptConn);
                            }
                        }
                    }
                    else if(events[i].events & EPOLLOUT)
                    {
                        acceptConn = (connection_t*) events[i].data.ptr;
                        if(acceptConn->fd == core_data.serverSocket.getSocketfd()) continue;
                        if(acceptConn == nullptr) continue;
                        writeEventHandler(acceptConn);
                    }
                }
            }
        }
        if(is_server_socket_received){
            conn_sock = core_data.serverSocket;
            while((accept_sock=conn_sock.Accept()) > 0){            
                //epoll et mode
                //从连接池获取空闲连接结构体
                connection_t* newConn = core_data.connections.getFreeConnection();
                newConn->setConnnection(accept_sock, READ_TYPE, nullptr, nullptr, false);
                newConn->addr = conn_sock.getAddr();
                newConn->addrlen = conn_sock.getAddrLen();
                Socket::setNonblockingSocket(accept_sock); //设为非阻塞socket
                temp = core_data.epoll.setEpollEvent(accept_sock, EPOLLIN | EPOLLET, (void*)newConn);
                core_data.epoll.epollAddEvent(temp, accept_sock);
            }
            if (conn_sock.getErrorno() != EAGAIN){
                this->status = -1;
                logger->critical("accept错误：{}", MyLogger::getSysError().c_str());
            }
        }
    }
}


void boss::readEventHandler(connection_t* connection)
{
    Socket clientSocket(connection->fd, CLIENT_SOCKET); 
    int ret;
    void* current = (void*)core_data.buffer;
    int bytes_read = 0;
    int bytes_left = max_buffer_size;
    auto logger = core_data.logger.setBasicLogger("boss");
    while(true){
        ret = clientSocket.Recv(current, MAX_RECV_SIZE);
        if(ret == 0){
            //用户已经断开连接
            core_data.epoll.epollDelEvent(connection->fd); //从epoll中移除事件
            clientSocket.Close();
            core_data.connections.returnConnection(connection);
            return;
        }else if(ret < 0){
            //发生错误
            if(clientSocket.getErrorno() == EAGAIN){ 
                break;
            }else{
                logger->critical("recv()失败，原因是：{}", MyLogger::getSysError().c_str()); 
                core_data.epoll.epollDelEvent(connection->fd);
                clientSocket.Close();
                core_data.connections.returnConnection(connection);
                return;
            }
        }else{
            //接收到客户端数据
            //协议解析
            if(connection->reader->protocolParser((char*) current, ret) == false){
                //用户发送数据有误，向客户端发送反馈信息
                json res;
                res["status"] = 404;
                res["msg"] = "协议解析错误，请检查您的数据格式";
                string data = res.dump();
                echo(data, connection, core_data);
                return;
            }else{
                if(connection->reader->is_complete){
                    //如果客户端完全发送协议数据，则根据协议内容执行下一步动作
                    break;
                }
            }
        } 
    }
    std::cout << "收到来自" << inet_ntoa(connection->addr.sin_addr) <<"的数据，数据为"<< connection->reader->data << std::endl;
    biz handler(core_data, connection);
    handler.run();
}

void boss::writeEventHandler(connection_t* conn)
{
    auto logger = core_data.logger.setBasicLogger("boss");
    int ret;
    if(conn == nullptr) return;
    responder* writer = conn->writer.get();
    if(writer == nullptr) return;
    Socket temp(writer->fd, CLIENT_SOCKET);
    {
        std::lock_guard<std::mutex>(writer->responder_lock); //获取responder_lock，防止线程竞争
        if(writer->bytes_left == -1 || writer->bytes_total == -1)
            return;
        while((ret = temp.Send(writer->current, writer->bytes_left)) > 0){
            writer->bytes_left -= ret;
            writer->current += ret;
            if(writer->bytes_left == 0)
                break;
        }
    }
    if(writer->bytes_left == 0)
    {
        core_data.epoll.epollDelEvent(conn->fd); //发送完毕，将写事件从epoll中移除
        conn->writer.reset(); //释放writer
    } 
    if(ret < 0){
        if(temp.getErrorno() != EAGAIN ){
            logger->error("send()失败，原因是：{}", MyLogger::getSysError().c_str());
        }
    }
    
}


void boss::run()
{
    std::thread bossThread(&boss::boss_cycle, this);
    bossThread.detach();
}